package org.dandsoft.blackbox.mapreduce.simple;

import org.dandsoft.blackbox.mapreduce.dispatcher.TaskDispatcher;
import org.dandsoft.blackbox.mapreduce.dispatcher.ProcessId;
import org.dandsoft.blackbox.mapreduce.dispatcher.ProcessCallbackHandler;
import org.dandsoft.blackbox.mapreduce.client.task.*;
import org.dandsoft.blackbox.mapreduce.processor.OperationProcessorDispatcher;
import org.dandsoft.blackbox.mapreduce.processor.OperationProcessor;
import org.dandsoft.blackbox.mapreduce.processor.OperationProcessorListener;

import java.util.Map;
import java.util.Hashtable;
import java.util.Set;

/**
 * @author Vilyuzhanin Denis (Fin)
 */
public class SimpleTaskDispatcher implements TaskDispatcher {

    private Map<ProcessId, ProcessInfo> processes = new Hashtable<ProcessId, ProcessInfo>();
    private OperationProcessorDispatcher processorDispatcher;
    private boolean isFinish;

    public synchronized ProcessId createProcess(Task task, ProcessCallbackHandler handler) {
        ProcessId pid = ProcessId.generate();
        ProcessInfo info = new ProcessInfo();
        info.task = task;
        info.pid = pid;
        info.queue = new ActionsQueue(task);
        info.handler = handler;
        processes.put(pid, info);
        return pid;
    }

    public synchronized void destoryProcess(ProcessId pid) {
        stop(pid);
        processes.remove(pid);
    }

    public void start(ProcessId pid) {
        final ProcessInfo info = processes.get(pid);
        synchronized (info.lock) {
            OperationProcessor processor = this.processorDispatcher.getProcessor(true, 0);
            try {
                while (!info.queue.isFinish()) {
                    Set<String> actions = info.queue.nextActions();
                    for (String action : actions) {
                        handle(pid, action, info, processor);
                    }
                    info.queue.setExecuted(actions);
                }
            } catch (Throwable e) {
                info.handler.processError(pid, this, e);
            } finally {
                this.processorDispatcher.releaseProcessor(processor);
            }
        }
    }

    private void handle(ProcessId pid, String action, final ProcessInfo processInfo, OperationProcessor processor) throws InterruptedException {
        if (ActionType.Input == processInfo.task.getActionType(action)) {
            //обрабатываем ввод данных
            Map map = processInfo.handler.handleInput(pid, this, action);
            processInfo.data.put(action, map);

        } else if (ActionType.Output == processInfo.task.getActionType(action)) {
            //обрабатываем вывод данных
            Set<String> dependencies = processInfo.task.getDependencies(action);
            for (String dep : dependencies) {
                Map data = processInfo.data.get(dep);
                processInfo.handler.handleOutput(pid, this, action, data);

            }
        } else if (ActionType.Register == processInfo.task.getActionType(action)) {
            RegisterInfo registerInfo = (RegisterInfo) processInfo.task.getActionInfo(action);
            //отчищаем хранилище
            //TODO: необходимо добавить транзакционнность
            registerInfo.getStorage().clear();
            Set<String> dependencies = processInfo.task.getDependencies(action);
            //сохраняем все данные в регистре
            for (String dep : dependencies) {
                Map data = processInfo.data.get(dep);
                registerInfo.getStorage().putAll(data);
            }
        } else if (ActionType.MapReduce == processInfo.task.getActionType(action)) {
            MapReduceInfo mapReduceInfo = (MapReduceInfo) processInfo.task.getActionInfo(action);
            for (String dep : processInfo.task.getDependencies(action)) {
                Map data = processInfo.data.get(dep);
                if (data != null) {
                    processor.input(data);
                }
            }
            OperationProcessorListener listener = new OperationProcessorListener() {
                public void success() {
                    synchronized (processInfo.lock) {
                        processInfo.lock.notifyAll();
                        isFinish = true;
                    }
                }

                public void error(Throwable t) {
                    synchronized (processInfo.lock) {
                        t.printStackTrace();
                        processInfo.lock.notifyAll();
                        isFinish = true;
                    }
                }
            };
            if (mapReduceInfo.getOperation() != null) {
                processor.setOperation(mapReduceInfo.getOperation());
            } else {
                processor.setOperationClass(mapReduceInfo.getOperationClass());
            }
            processor.addListener(listener);
            processor.execute();
            if (!isFinish) {
                processInfo.lock.wait();
            }
            isFinish = false;
            Map map = processor.output();
            processInfo.data.put(action, map);
            processor.reset();
        }
    }


    public void stop(ProcessId pid) {

    }


    static class ProcessInfo {
        public final Object lock = new Object();
        public ProcessId pid;
        public Task task;
        public ActionsQueue queue;
        public ProcessCallbackHandler handler;
        public Map<String, Map> data = new Hashtable<String, Map>();
    }

    public OperationProcessorDispatcher getProcessorDispatcher() {
        return processorDispatcher;
    }

    public void setProcessorDispatcher(OperationProcessorDispatcher processorDispatcher) {
        this.processorDispatcher = processorDispatcher;
    }
}
